package org.example;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Arrays;
import java.util.Properties;

public class Producer {
    private KafkaProducer<String,String> kafkaProducer;
    Producer(){
        //1.创建kafka生产者的配置对象
        Properties properties = new Properties();
//        虚拟机本地
//        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.10.222:9092");
        //本地端口
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
//        服务器
//        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"101.42.19.16:9092");
        //序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //创建kafka生产者对象
        kafkaProducer = new KafkaProducer<>(properties);

    }

    public void send(String[] list,int partition){
        String s = Arrays.toString(list);
        s  = s.replace("[","");
        s  =s.replace("]","");
        s = s.replaceAll(", "," ");

        //发送
        kafkaProducer.send(new ProducerRecord<>("temp",partition,null,s));

    }
    /*
    * 关闭
    * */
    public void stop(){
        kafkaProducer.close();
    }
}
